[Rust] Testcontainerで使い捨てコンテナを使ったテスト

[Rust] Testcontainerで使い捨てコンテナを使ったテスト

#Rust
#Elasticsearch
中村 修太

中村 修太

facebook logohatena logotwitter logo
Clock Icon2024.11.05

Introduction

Testcontainers は、
データストアやメッセージブローカーなど、Dockerコンテナで実行可能なほぼすべてのサービスに対して
使い捨ての軽量コンテナをAPIで提供するライブラリです。
実際のサービス・ソフトウェアが対象マシンにインストールされていなくても、
実行時にコンテナを作成して実行させることができます。

例えば、Redisを使用するシステムがあったとします。
ローカルでRedisにアクセスするunit testがある場合、
Redisをインストールしなくても、Testcontainersをunit test時に使用すれば、
テスト実行時に使い捨てのコンテナを起動してRedisをインストールし、テストすることができます。

Testcontainersはさまざまな言語で使用することができます。
(Java,Go,Node.js,Pythonなどなど）
今回はRustからTestcontainersをつかってみます。

Testcontainers

TestcontainersはコードからTestcontainers APIを実行し、
Docker EngineでRyukコンテナとAPIで指定されたコンテナを起動します。
※RyukコンテナはTestcontainersのライフサイクル管理をする

テスト用コンテナが起動したらテストを実施し、
終了したらクリーンアップします。

Testcontainersを使うにはTestcontainers Cloudを使うか
Docker API 互換のコンテナ ランタイムが必要です。
対応しているランタイムはここから参照してください。

なお、今回はcolima + Docker Engineで試してみました。

Environment

  • MacBook Pro (14-inch, M3, 2023)
  • OS : MacOS 14.5
  • colima : 0.7.6
  • docker : 27.3.1

Setup

コンテナランタイムをインストールしましょう。
今回はcolimaとdockerをインストールします。

% brew install colima
% brew install docker

colimaを起動します。
また、テスト実行時に↓のようなエラーがでる場合があります。

Error: Client(Init(SocketNotFoundError("/var/run/docker.sock")))

必要に応じてDOCKER_HOST環境変数を設定してください。

% colima start

% export DOCKER_HOST="unix://${HOME}/.colima/default/docker.sock"

Try

では試しに、ElasticSearchをつかったテストを実装してみましょう。
Cargoでプロジェクトを作成して必要な依存関係を記述します。

% cargo new testcontainers-example
% cd testcontainers-example

#Cargo.toml
[dependencies]
anyhow = "1.0.91"
elasticsearch = "8.15.0-alpha.1"
serde = { version = "1.0.214", features = ["derive"] }
serde_json = "1.0.132"
sqlx = { version = "0.8.2", features = ["runtime-tokio-native-tls", "postgres"] }
testcontainers = "0.23.1"
tokio = { version = "1.41.0", features = ["full"] }
reqwest = { version = "0.11", features = ["json"] }
tracing = "0.1.40"
tracing-subscriber = "0.3.18"

↓のこのコードは、Elasticsearchを操作する簡単な実装と、testcontainersをつかったテストコードです。

//main.rs
use elasticsearch::{
    Elasticsearch, 
    http::transport::Transport,  
    indices::{IndicesCreateParts, IndicesDeleteParts}, 
    params::Refresh, 
    IndexParts, 
};
// シリアライズ/デシリアライズ用のトレイト
use serde::{Deserialize, Serialize};
use serde_json::Value;  // JSON操作用
use std::error::Error;

/// Elasticsearchに保存するドキュメントの構造体
#[derive(Serialize, Deserialize, Debug, PartialEq)]
pub struct Document {
    pub id: String,      // ドキュメントのID
    pub title: String,   // タイトル
    pub content: String, // 内容
}

/// Elasticsearchとの通信を担当するリポジトリ
pub struct ElasticsearchRepository {
    client: Elasticsearch,  // Elasticsearchクライアントのインスタンス
}

impl ElasticsearchRepository {
    /// 新しいリポジトリインスタンスを作成
    pub fn new(client: Elasticsearch) -> Self {
        Self { client }
    }

    /// インデックスを作成する
    /// 
    /// # Arguments
    /// * `index_name` - 作成するインデックスの名前
    pub async fn create_index(&self, index_name: &str) -> Result<(), Box<dyn Error>> {
        self.client
            .indices()
            .create(IndicesCreateParts::Index(index_name))
            .send()
            .await?;
        Ok(())
    }

    /// インデックスを削除する
    /// 
    /// # Arguments
    /// * `index_name` - 削除するインデックスの名前
    pub async fn delete_index(&self, index_name: &str) -> Result<(), Box<dyn Error>> {
        self.client
            .indices()
            .delete(IndicesDeleteParts::Index(&[index_name]))
            .send()
            .await?;
        Ok(())
    }

    /// ドキュメントをインデックスに追加する
    /// 
    /// # Arguments
    /// * `index_name` - 対象のインデックス名
    /// * `doc` - 追加するドキュメント
    pub async fn index_document(&self, index_name: &str, doc: &Document) -> Result<(), Box<dyn Error>> {
        self.client
            .index(IndexParts::IndexId(index_name, &doc.id))  // インデックスとIDを指定
            .body(doc)  // ドキュメントの内容
            .refresh(Refresh::True)  // 即時検索可能にする
            .send()
            .await?;
        Ok(())
    }

    /// ドキュメントを取得する
    /// 
    /// # Arguments
    /// * `index_name` - 対象のインデックス名
    /// * `id` - 取得するドキュメントのID
    /// 
    /// # Returns
    /// * `Ok(Some(Document))` - ドキュメントが見つかった場合
    /// * `Ok(None)` - ドキュメントが見つからなかった場合
    pub async fn get_document(&self, index_name: &str, id: &str) -> Result<Option<Document>, Box<dyn Error>> {
        let response = self.client
            .get(elasticsearch::GetParts::IndexId(index_name, id))
            .send()
            .await?;

        if response.status_code().is_success() {
            let response_body: Value = response.json().await?;
            // ドキュメントが見つかったかチェック
            if response_body["found"].as_bool().unwrap_or(false) {
                // ドキュメントの内容を取得
                if let Some(source) = response_body["_source"].as_object() {
                    let doc: Document = serde_json::from_value(source.clone().into())?;
                    Ok(Some(doc))
                } else {
                    Ok(None)
                }
            } else {
                Ok(None)
            }
        } else {
            Ok(None)
        }
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use std::time::Duration;
    use testcontainers::{
        core::{ContainerPort, WaitFor},
        runners::AsyncRunner,
        GenericImage,
        ImageExt,
    };
    use reqwest::Client;
    // ロギング用のクレートをインポート
    use tracing::{info, debug, error};

    // Elasticsearchのデフォルトポート
    const ES_PORT: u16 = 9200;

    /// Elasticsearchサーバーが利用可能になるまで待機する
    /// 
    /// # Arguments
    /// * `url` - ElasticsearchサーバーのURL
    /// 
    /// # Returns
    /// * `Ok(())` - サーバーが利用可能になった場合
    /// * `Err` - タイムアウトまたはエラーが発生した場合
    /// 
    async fn wait_for_elasticsearch(url: &str) -> Result<(), Box<dyn Error>> {
        let client = Client::new();
        let health_url = format!("{url}/_cluster/health");

        for i in 0..12 {
            debug!("Attempt {} to connect to Elasticsearch", i + 1);
            match client.get(&health_url).send().await {
                Ok(response) if response.status().is_success() => {
                    info!("Elasticsearch is ready!");
                    return Ok(());
                }
                Ok(response) => {
                    debug!("Elasticsearch not ready yet. Status: {}", response.status());
                    tokio::time::sleep(Duration::from_secs(5)).await;
                }
                Err(e) => {
                    error!("Error connecting to Elasticsearch: {}", e);
                    tokio::time::sleep(Duration::from_secs(5)).await;
                }
            }
        }

        error!("Elasticsearch failed to become ready after all attempts");
        Err("Elasticsearch failed to become ready".into())
    }

    /// ElasticsearchRepositoryのテスト
    #[tokio::test]
    async fn test_elasticsearch_repository() -> Result<(), Box<dyn Error>> {
        let subscriber = tracing_subscriber::fmt()
            .with_max_level(tracing::Level::DEBUG)
            .with_test_writer()
            .init();

        info!("Starting Elasticsearch container test");

        // Elasticsearchコンテナの設定と起動
        let container = GenericImage::new("elasticsearch", "8.12.0")
            .with_exposed_port(ContainerPort::Tcp(ES_PORT))
            .with_wait_for(WaitFor::message_on_stdout("started"))
            // シングルノードモードで実行（テスト用）
            .with_env_var("discovery.type", "single-node")
            // セキュリティ機能を無効化（テスト用）
            .with_env_var("xpack.security.enabled", "false")
            .with_env_var("ES_JAVA_OPTS", "-Xms512m -Xmx512m")
            // コンテナ起動のタイムアウトを設定
            .with_startup_timeout(Duration::from_secs(90))
            .start()
            .await?;

        // コンテナのホストとポートを取得
        let host = container.get_host().await?;
        let host_port = container.get_host_port_ipv4(ES_PORT).await?;

        // ElasticsearchのURLを構築
        let url = format!("http://{}:{}", host, host_port);
        info!("Elasticsearch URL: {}", url);

        debug!("Waiting for Elasticsearch to be ready...");
        wait_for_elasticsearch(&url).await?;

        // Elasticsearchクライアントとリポジトリの初期化
        let transport = Transport::single_node(&url)?;
        let client = Elasticsearch::new(transport);
        let repo = ElasticsearchRepository::new(client);

        let index_name = "test_index";
        debug!("Creating index: {}", index_name);
        repo.create_index(index_name).await?;

        tokio::time::sleep(Duration::from_secs(2)).await;

        let test_doc = Document {
            id: "1".to_string(),
            title: "Test Title".to_string(),
            content: "Test Content".to_string(),
        };

        debug!("Indexing document: {:?}", test_doc);
        repo.index_document(index_name, &test_doc).await?;

        tokio::time::sleep(Duration::from_secs(2)).await;

        debug!("Retrieving document with id: {}", test_doc.id);
        let retrieved_doc = repo.get_document(index_name, "1").await?;
        info!("Retrieved document: {:?}", retrieved_doc);
        assert!(retrieved_doc.is_some());
        assert_eq!(retrieved_doc.unwrap(), test_doc);

        debug!("Testing non-existent document retrieval");
        let non_existent = repo.get_document(index_name, "999").await?;
        assert!(non_existent.is_none());

        debug!("Cleaning up: deleting index {}", index_name);
        repo.delete_index(index_name).await?;

        info!("Test completed successfully");
        Ok(())
    }
}

ElasticsearchRepositoryでは、Elasticsearchと通信を行うための簡単な処理を実装しています。

let container = GenericImage::new("elasticsearch", "8.12.0")
    .with_exposed_port(ContainerPort::Tcp(ES_PORT))
    .with_wait_for(WaitFor::message_on_stdout("started"))
    .with_env_var("discovery.type", "single-node")
    .with_env_var("xpack.security.enabled", "false")
    .with_env_var("ES_JAVA_OPTS", "-Xms512m -Xmx512m")
    .with_startup_timeout(Duration::from_secs(90))
    .start()
    .await?;

GenericImageをつかってイメージを詳細に指定しています。
バージョンやポートを指定し、環境変数も設定しています。
もちろん自作のDockerイメージを使うこともできます。

async fn wait_for_elasticsearch(url: &str) -> Result<(), Box<dyn Error>> {
    for i in 0..12 {
        match client.get(&health_url).send().await {
            Ok(response) if response.status().is_success() => return Ok(()),
            // エラー処理...
        }
    }
}

wait_for_elasticsearchでは、Elasticsearchの準備完了するまで待機します。
あとはElasticsearchを使ったテストを実行。

テストを実行すると下記のようにパスします。
（コンテナ起動とかでけっこう時間がかかる）
ElasticSearchがローカルにインストールしていなくても動作します。

% cargo test

・・・

running 1 test
test tests::test_elasticsearch_repository ... ok

test result: ok. 1 passed; 0 failed; 0 ignored; 0 measured; 0 filtered out; finished in 19.79s

コンテナは使い捨てなので、実行するたびにまっさらな状態で実行できます。

Summary

今回はTestcontainersで使い捨てコンテナをつかったunit testを試して見ました。
ローカルでテストするときはもちろん、CIを使うときも役立ちそうです。

References

Share this article

facebook logohatena logotwitter logo

関連記事

[Rust] Amazon Transcribe Streamingを使ってmp3をテキスト化する

[Rust] Amazon Transcribe Streamingを使ってmp3をテキスト化する

User avatar
中村 修太
2024.10.24
axum on LambdaでHTTPレスポンスをストリーミング配信(Server-Sent Events)する

axum on LambdaでHTTPレスポンスをストリーミング配信(Server-Sent Events)する

User avatar
shuntaka
2024.10.20
RustでSMTPプロトコルを使ってメール送信してみた

RustでSMTPプロトコルを使ってメール送信してみた

User avatar
morimorikochan
2024.10.16
Rustで暗号利用モードをいろいろ実装してみた

Rustで暗号利用モードをいろいろ実装してみた

User avatar
morimorikochan
2024.10.07
Classmethod's white logo
Facebook's logo
X's logo
YouTube's logo

© Classmethod, Inc. All rights reserved.